package com.webmining.sherlock.store;

import com.webmining.sherlock.admin.Configuration;
import com.webmining.sherlock.admin.ConfigurationException;
import com.webmining.sherlock.search.Search;
import com.webmining.sherlock.util.search.IOConverter;
import org.apache.log4j.Logger;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.ru.RussianAnalyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.IndexWriterConfig;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.RAMDirectory;
import org.apache.lucene.util.Version;
import org.hsqldb.util.SqlFile;
import org.hsqldb.util.SqlToolError;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.sql.*;
import java.util.HashMap;
import java.util.Hashtable;


/**
 * Connection to the sherlock store
 */
public class Store {
	
	/**
	 * Internal reference to a lucene index
	 */
    protected IndexWriter indexWriter;

    /**
     * Internal native connection object
     */
    private Connection connection;
    
    /**
     * opened database flag
     */
    private boolean databaseOpened;
    
    /**
     * opened Apache Lucene index flag
     */
    private boolean indexOpened;
    
    /**
     * transaction database flag
     */
    private boolean transactionOpened;
    
    /**
     * SQL sentence for query a sequence for primary key generation
     */
    private String sequenceQuery;
    
    /**
     * Logger
     */
    protected Logger log;

    /**
     * this flag will be active if the database native sequence
     * code generation fails
     */
    boolean useInternalSequence = false;

    /**
     * internal generation code sequence store
     */
    Hashtable lastCodes = new Hashtable();
    private Directory indexDirectory;

    /**
     * Creates a new store with a closed database connection
     */
    public Store() {
    	log = Logger.getLogger(Store.class);
        connection = null;
        databaseOpened = false;
        indexOpened = false;
    }

    /**
     * Access the Sherlock index and opens internal database connection
     */
    public void open() throws ConfigurationException, SQLException, IOException {
        if (!indexOpened) {
        	// Open the document index
            Analyzer analyzer = new RussianAnalyzer(Version.LUCENE_45);
            File index = new File("store/index");
            if (!index.exists()){
                index.mkdirs();
                System.out.println(index.getAbsolutePath());
            }
            System.out.println(index.getAbsolutePath());
            indexDirectory = new RAMDirectory();
            Search.setIndex(indexDirectory);
            IndexWriterConfig config = new IndexWriterConfig(Version.LUCENE_45, analyzer);

        	indexWriter = new IndexWriter(indexDirectory,config);
        	indexOpened = true;
        	log.debug("index " + indexDirectory + " opened for sherlock store");
        }
        if (!databaseOpened) {
        	// Initialize internal sequence generation resources in
        	// case the database has no internal sequence support
            useInternalSequence = false;
            Hashtable lastCodes = new Hashtable();
        	// Open the database connection
            String url = Configuration.getStoreURL();
            String classDriver = Configuration.getStoreClassDriver();
            String user = Configuration.getStoreUser();
            String password = Configuration.getStorePassword();
            String catalog = Configuration.getStoreCatalog();
            sequenceQuery = Configuration.getStoreSequenceQuery();
            try {
            	Class.forName(classDriver);
            } catch(ClassNotFoundException cnfe) {
            	log.error("Unable to load the database class driver " +
            			classDriver, cnfe);
            	throw new SQLException("Unable to load the database class " +
            			"driver " + classDriver + " : " + cnfe.getMessage());
            }
            connection = DriverManager.getConnection(url, user, password);
            if (catalog != null && !catalog.trim().equals("")) {
            	connection.setCatalog(catalog);
            } else {
        		log.debug("Default catalog selected for sherlock store");
            }
            databaseOpened = true;
            log.debug("database " + url + " opened for sherlock store");
        }
    }

    public void openIndex(){
        try {
            Analyzer analyzer = new RussianAnalyzer(Version.LUCENE_45);
            IndexWriterConfig config = new IndexWriterConfig(Version.LUCENE_45, analyzer);
            indexWriter = new IndexWriter(indexDirectory,config);
        } catch (IOException e) {
            e.printStackTrace();
        }
        indexOpened = true;
    }

    public void closeIndex(){
        if (indexOpened) {
            try {
                indexWriter.close();
                indexOpened = false;
            } catch (IOException ioe){
                log.warn("Unable to close index", ioe);
            }
        }
    }

    /**
     * Close the internal database connection
     */
    public void close() {
    	log.debug("closing store");
        if (indexOpened) {
        	try {
        		indexWriter.close();
        		indexOpened = false;
        	} catch (IOException ioe){
            	log.warn("Unable to close index", ioe);
        	}
        }
        if (databaseOpened) {
            try{
            	if ("internal".equals(Configuration.getStoreType())) {
            		// do not issue a SHUTDOWN for
            		// internal database
            		// TODO: check if other extra close operations are
            		// need for internal database
            	}
                if (connection != null) {
                    connection.close();
                    connection = null;
                }
                databaseOpened = false;
            } catch(ConfigurationException ce) {
            	log.warn("Unable to determine store type for closing", ce);
            } catch(SQLException se) {
            	log.warn("Unable to close connection", se);
            }
        }
    }

    /**
     * Check if a valid database is opened
     */
    public void checkValidConnection() throws SQLException {
        if (connection == null || connection.isClosed()) {
            throw new SQLException("Connection no opened");
        }
    }
    
    /**
     * Creates a statement
     * @return Statement created
     */
    public Statement createStatement() throws SQLException {
        checkValidConnection();
        return connection.createStatement();
    }
    
    /**
     * Opens a transaction
     */
    public void beginTransaction() throws SQLException {
        checkValidConnection();
        if (transactionOpened) {
            throw new SQLException ("Attempt to open transaction twice");
        } else {
            connection.setAutoCommit(false);
            transactionOpened = true;
        }
    }
    
    /**
     * Close a transaction
     * @throws SQLException for a potential database error
     */
    public void commitTransaction() throws SQLException {
        checkValidConnection();
        if (transactionOpened) {
            connection.commit();
            connection.setAutoCommit(true);
            transactionOpened = false;
        } else {
            throw new SQLException("Transaction no opened");
        }
    }
    
    /**
     * Aborts a transaction.
     * @throws SQLException for a potential database error
     */
    public void abortTransaction() throws SQLException {
        checkValidConnection();
        if (transactionOpened) {
            connection.rollback();
            connection.setAutoCommit(true);
            setTransactionOpened(false);
        } else {
            throw new SQLException ("Transaction no opened");
        }
    }
    
    /**
     * Generates a sequential code for a given tables. For this generation to
     * work the <code>sherlock.store.sequencequery</code> property in the sherlock
     * configuration file should be set properly.
     * @param component name to generate its sequential code
     * @return The next code generated for the component's table
     */
    public int nextCode(String component) throws SQLException{
        int code = 0;
        // try to get a next code based on database sequences
        if (!useInternalSequence) {
	        try {
	        	// The sequence query has the ? character as a parameter
	        	// to set the component name, so replace it adequately
	        	PreparedStatement sentence = prepareStatement(
	        			sequenceQuery.replaceAll("\\?", component));
	        	ResultSet result = sentence.executeQuery();
	        	if (result.next()) {
	        		// Get the next code from the first column
	        		code = result.getInt(1);
	        	} else {
	        		log.error("Unable to generate the next code for component " +
	        				component);
	        		throw new SQLException("Unable to generate the next code for " +
	        				"component " + component);
	        	}
	        } catch (SQLException se) {
	        	log.warn("unable to access database sequences, use internal sequences", se);
	        	useInternalSequence=true;
	        }
        }
        // if the a use internal sequence was signaled from a previous next code generation
        // or from the current next code generation, perform a next internal code generation
        if (useInternalSequence) {
        	code = nextInternalCode(component);
        }
        return code;
    }
    
    synchronized public int nextInternalCode(String component) throws SQLException {
    	int code = 0;
    	Integer lastCode = (Integer) lastCodes.get(component);
    	// if no last code given for component, select the max code as current code
    	if (lastCode == null) {
    		beginTransaction();
        	PreparedStatement sentence = prepareStatement(
        			"select max(code_" + component + ") from ox_" + component);
        	ResultSet result = sentence.executeQuery();
        	if (result.next()) {
        		// Get the next code from the first column
        		code = result.getInt(1) + 1;
        	}
        	commitTransaction();
    	} else {
    		code = lastCode.intValue() + 1;
    	}
    	lastCodes.put(component, new Integer(code));
    	return code;
    }
    
    public PreparedStatement prepareStatement(String sentence)
    		throws SQLException {
    	return connection.prepareStatement(sentence);
    }
    
    /**
     * Executes and update sentence and close the statement
     * @param sql The SQL sentence to execute
     * @return Number of rows affected
     */
    public int executeUpdate(String sql) throws SQLException {
        Statement sentence = createStatement();
        if (sentence == null) {
            throw new SQLException("Unable to create the sentence");
        }
        int rowsAffected = sentence.executeUpdate(sql);
        sentence.close();
        sentence = null;
        return rowsAffected;
    }
    
    /**
     * Close the database connection
     * @throws Throwable for a potential general error
     */
    protected void finalize() throws Throwable {
    	if (databaseOpened || indexOpened) {
    		log.warn("the store should not be closed on finalize");
    		close();
    	}
    }
    
    /**
     * Reads if the transaction is opened
     * @return internal flag value
     */
    public boolean isTransactionOpened() {
        return transactionOpened;
    }
    
    /**
     * Sets the internal transaction state
     * @param transactionOpened new flag value
     */
    private void setTransactionOpened(boolean transactionOpened) {
        this.transactionOpened = transactionOpened;
    }

	public void addDocument(Document doc) throws IOException {
        openIndex();
		indexWriter.addDocument(doc);
        closeIndex();
	}

	public static void verifyObjects() throws ConfigurationException,
			IOException, SQLException {
		if ("internal".equals(Configuration.getStoreType())) {
			Store store = new Store();
			store.open();
			Statement statement = store.createStatement();
			try {
				statement.execute("select 1 from ox_page");
			} catch (SQLException se) {
				// if no object exists create new sherlock objects
				// in the database
                InputStream is = Store.class.getClassLoader()
                        .getResourceAsStream(Configuration.getSuppportDirectory() +
                                "/" + "hsqldb" + "/" + "sherlock-hsqldb.sql");



				File sqlScriptFile = IOConverter.inputStreamToFile(is);
				SqlFile sqlScript = new SqlFile(sqlScriptFile, false, new HashMap());
				try {
					sqlScript.execute(store.connection, false);
				} catch (SQLException sqle) {
					Logger log = Logger.getLogger(Store.class);
					log.warn("Sherlock internal database objects were not successfully " +
							"created. This is normal if the exception reports sequence " +
							"creation.", sqle);
				} catch (SqlToolError ste) {
					Logger log = Logger.getLogger(Store.class);
					log.error("Unable to call database script generation", ste);
					throw new SQLException("Unable to call database script generation");
				}
			} finally {
				// ensure to close the store after verification
				store.close();
			}
		}
		
	}
	
}
